<!DOCTYPE html>
<html xmlns="http://www.w3.org/1999/xhtml">
  <head>

    <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
    <meta content="Cask Data, Inc." name="author" />
<meta content="Copyright © 2014-2016 Cask Data, Inc." name="copyright" />


    <meta name="git_release" content="6.1.1">
    <meta name="git_hash" content="05fbac36f9f7aadeb44f5728cea35136dbc243e5">
    <meta name="git_timestamp" content="2020-02-09 08:22:47 +0800">
    <title>Batch Data Processing with CDAP</title>

    <link rel="stylesheet" href="../_static/cdap-bootstrap.css" type="text/css" />
    <link rel="stylesheet" href="../_static/pygments.css" type="text/css" />
    <link rel="stylesheet" href="../_static/bootstrap-3.3.6/css/bootstrap.min.css" type="text/css" />
    <link rel="stylesheet" href="../_static/bootstrap-3.3.6/css/bootstrap-theme.min.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/bootstrap-sphinx.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/cdap-dynamicscrollspy-4.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/jquery.mCustomScrollbar.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/cdap-jquery.mCustomScrollbar.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/abixTreeList-2.css" type="text/css" />
    <link rel="stylesheet" href="../_static/cdap-bootstrap.css" type="text/css" />

    <script type="text/javascript">
      var DOCUMENTATION_OPTIONS = {
        URL_ROOT:    '',
        VERSION:     '6.1.1',
        COLLAPSE_INDEX: false,
        FILE_SUFFIX: '.html',
        HAS_SOURCE:  false
      };
    </script>
    <script type="text/javascript" src="../_static/jquery.js"></script>
    <script type="text/javascript" src="../_static/underscore.js"></script>
    <script type="text/javascript" src="../_static/doctools.js"></script>
    <script type="text/javascript" src="../_static/language_data.js"></script>

    <link rel="shortcut icon" href="../_static/favicon.ico"/>
    <link rel="index" title="Index" href="../genindex.html" />
    <link rel="search" title="Search" href="../search.html" />
    <link rel="top" title="Cask Data Application Platform 6.1.1 Documentation" href="../index.html" />
    <link rel="up" title="How-To Guides" href="index.html" />
    <link rel="next" title="Iterative Data Processing with Apache Spark" href="cdap-spark-guide.html" />
    <link rel="prev" title="Consuming Data from Kafka" href="cdap-kafka-ingest-guide.html" />
    <!-- block extrahead -->
    <meta charset='utf-8'>
    <meta http-equiv='X-UA-Compatible' content='IE=edge,chrome=1'>
    <meta name='viewport' content='width=device-width, initial-scale=1.0, maximum-scale=1'>
    <meta name="apple-mobile-web-app-capable" content="yes">
    <!-- block extrahead end -->

</head>
<body role="document">

<!-- block navbar -->
<div id="navbar" class="navbar navbar-inverse navbar-default navbar-fixed-top">
    <div class="container-fluid">
      <div class="row">
        <div class="navbar-header">
          <!-- .btn-navbar is used as the toggle for collapsed navbar content -->
          <a class="navbar-brand" href="../table-of-contents/../../index.html">
            <span><img alt="CDAP logo" src="../_static/cdap_logo.svg"/></span>
          </a>

          <button type="button" class="navbar-toggle" data-toggle="collapse" data-target=".nav-collapse">
            <span class="icon-bar"></span>
            <span class="icon-bar"></span>
            <span class="icon-bar"></span>
          </button>

          <div class="pull-right">
            <div class="dropdown version-dropdown">
              <a href="#" class="dropdown-toggle" data-toggle="dropdown"
                role="button" aria-haspopup="true" aria-expanded="false">
                v 6.1.1 <span class="caret"></span>
              </a>
              <ul class="dropdown-menu">
                <li><a href="//docs.cdap.io/cdap/5.1.2/en/index.html">v 5.1.2</a></li>
                <li><a href="//docs.cdap.io/cdap/4.3.4/en/index.html">v 4.3.4</a></li>
              </ul>
            </div>
          </div>
          <form class="navbar-form navbar-right navbar-search" action="../search.html" method="get">
            <div class="form-group">
              <div class="navbar-search-image material-icons"></div>
              <input type="text" name="q" class="form-control" placeholder="  Search" />
            </div>
            <input type="hidden" name="check_keywords" value="yes" />
            <input type="hidden" name="area" value="default" />
          </form>

          <div class="collapse navbar-collapse nav-collapse navbar-right navbar-navigation">
            <ul class="nav navbar-nav"><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../table-of-contents/../../index.html">简介</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link current" href="../table-of-contents/../../guides.html">手册</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../table-of-contents/../../reference-manual/index.html">参考</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../table-of-contents/../../faqs/index.html">帮助</a></li>
            </ul>
          </div>

        </div>
      </div>
    </div>
  </div><!-- block navbar end -->
<!-- block main content -->
<div class="main-container container">
  <div class="row"><div class="col-md-2">
      <div id="sidebar" class="bs-sidenav scrollable-y-outside" role="complementary">
<!-- theme_manual: examples-manual -->
<!-- theme_manual_highlight: guides -->
<!-- sidebar_title_link: ../table-of-contents/../../guides.html -->

  <div role="note" aria-label="manuals links"><h3><a href="../table-of-contents/../../guides.html">Guides</a></h3>

    <ul class="this-page-menu">
      <li class="toctree-l1"><a href="../table-of-contents/../../user-guide/index.html" rel="nofollow">用户手册</a>
      </li>
      <li class="toctree-l1"><a href="../table-of-contents/../../developer-manual/index.html" rel="nofollow">开发手册</a>
      </li>
      <li class="toctree-l1"><a href="../table-of-contents/../../admin-manual/index.html" rel="nofollow">管理手册</a>
      </li>
      <li class="toctree-l1"><a href="../table-of-contents/../../integrations/index.html" rel="nofollow">集成手册</a>
      </li>
      <li class="toctree-l1"><b><a href="../table-of-contents/../../examples-manual/index.html" rel="nofollow">最佳实践</a></b>
      <nav class="pagenav">
      <ul class="current">
<li class="toctree-l1"><a class="reference internal" href="../index.html">Introduction</a></li>
<li class="toctree-l1 current"><a class="reference internal" href="index.html">How-To Guides</a><ul class="current">
<li class="toctree-l2"><a class="reference internal" href="cdap-bi-guide.html">Analyzing CDAP Data from BI Tools</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-cube-guide.html">Data Analysis with OLAP Cube</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-etl-guide.html">Creating Hydrator Applications using CDAP System Artifacts</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-flow-guide.html">Real-Time Data Processing with a Flow</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-flume-guide.html">Ingesting Data into CDAP using Apache Flume</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-kafka-ingest-guide.html">Consuming Data from Kafka</a></li>
<li class="toctree-l2 current"><a class="current reference internal" href="#">Batch Data Processing with CDAP</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-spark-guide.html">Iterative Data Processing with Apache Spark</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-timeseries-guide.html">Storing Timeseries Data</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-twitter-ingest-guide.html">Consuming Twitter Data in Real Time</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-workflow-guide.html">Batch Data Processing with CDAP using Workflow</a></li>
</ul>
</li>
</ul>
</nav>
      </li>
    </ul>
  </div></div>
    </div><div class="col-md-8 content" id="main-content">
    
  <div class="section" id="batch-data-processing-with-cdap">
<span id="cdap-mapreduce-guide"></span><h1>Batch Data Processing with CDAP<a class="headerlink" href="#batch-data-processing-with-cdap" title="Permalink to this headline">🔗</a></h1>
<blockquote class="pull-quote">
<div><strong>Source Code Repository:</strong> Source code (and other resources) for this guide are available at the
<a class="reference external" href="https://github.com/cdap-guides/cdap-mapreduce-guide/tree/develop">CDAP Guides GitHub repository</a>.</div></blockquote>
<p><a class="reference external" href="http://research.google.com/archive/mapreduce.html">MapReduce</a> is the
most popular paradigm for processing large amounts of data in a reliable
and fault-tolerant manner. In this guide, you will learn how to batch
process data using MapReduce in the <a class="reference external" href="http://cdap.io">Cask Data Application Platform
(CDAP)</a>.</p>
<div class="section" id="what-you-will-build">
<h2>What You Will Build<a class="headerlink" href="#what-you-will-build" title="Permalink to this headline">🔗</a></h2>
<p>This guide will take you through building a
<a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/applications.html">CDAP application</a>
that uses ingested Apache access log events to compute the top 10 client IPs in a
specific time-range and query the results. You will:</p>
<ul class="simple">
<li>Build a
<a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/mapreduce-jobs.html">MapReduce program</a>
to process Apache access log events;</li>
<li>Use a
<a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/datasets/index.html">Dataset</a>
to persist results of the MapReduce program; and</li>
<li>Build a
<a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/services.html">Service</a>
to serve the results via HTTP.</li>
</ul>
</div>
<div class="section" id="what-you-will-need">
<h2>What You Will Need<a class="headerlink" href="#what-you-will-need" title="Permalink to this headline">🔗</a></h2>
<ul class="simple">
<li><a class="reference external" href="http://www.oracle.com/technetwork/java/javase/downloads/index.html">JDK 7 or 8</a></li>
<li><a class="reference external" href="http://maven.apache.org/">Apache Maven 3.1+</a></li>
<li><a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/getting-started/standalone/index.html">CDAP SDK</a></li>
</ul>
</div>
<div class="section" id="lets-build-it">
<h2>Let’s Build It!<a class="headerlink" href="#lets-build-it" title="Permalink to this headline">🔗</a></h2>
<p>The following sections will guide you through building an application from scratch. If you
are interested in deploying and running the application right away, you can clone its
source code from this GitHub repository. In that case, feel free to skip the next two
sections and jump right to the
<a class="reference external" href="#build-and-run-application">Build and Run Application</a> section.</p>
<div class="section" id="application-design">
<h3>Application Design<a class="headerlink" href="#application-design" title="Permalink to this headline">🔗</a></h3>
<p>The application will assume that the Apache access logs are ingested
into a Stream. The log events can be ingested into a Stream continuously
in real-time or in batches; whichever way, it doesn’t affect the ability
of the MapReduce program to consume them.</p>
<p>The MapReduce program extracts the required information from the raw logs
and computes the top 10 Client IPs by traffic in a specific time range.
The results of the computation are persisted in a Dataset.</p>
<p>Finally, the application contains a Service that exposes an HTTP
endpoint to access the data stored in the Dataset.</p>
<a class="reference internal image-reference" href="../_images/app-design5.png"><img alt="../_images/app-design5.png" class="align-center" src="../_images/app-design5.png" style="width: 8in;" /></a>
</div>
<div class="section" id="implementation">
<h3>Implementation<a class="headerlink" href="#implementation" title="Permalink to this headline">🔗</a></h3>
<p>The first step is to construct our application structure. We will use a
standard Maven project structure for all of the source code files:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="go">./pom.xml</span>
<span class="go">./src/main/java/co/cdap/guides/ClientCount.java</span>
<span class="go">./src/main/java/co/cdap/guides/CountsCombiner.java</span>
<span class="go">./src/main/java/co/cdap/guides/IPMapper.java</span>
<span class="go">./src/main/java/co/cdap/guides/LogAnalyticsApp.java</span>
<span class="go">./src/main/java/co/cdap/guides/TopClientsMapReduce.java</span>
<span class="go">./src/main/java/co/cdap/guides/TopClientsService.java</span>
<span class="go">./src/main/java/co/cdap/guides/TopNClientsReducer.java</span>
</pre></div>
</div>
<p>The CDAP application is identified by the <code class="docutils literal notranslate"><span class="pre">LogAnalyticsApp</span></code> class. This
class extends an <a class="reference external" href="http://docs.cdap.io/cdap/current/en/reference-manual/javadocs/co/cask/cdap/api/app/AbstractApplication.html">AbstractApplication</a>,
and overrides the <code class="docutils literal notranslate"><span class="pre">configure()</span></code> method to define all of the application components:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">LogAnalyticsApp</span> <span class="kd">extends</span> <span class="n">AbstractApplication</span> <span class="p">{</span>

  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">DATASET_NAME</span> <span class="o">=</span> <span class="s">&quot;topClients&quot;</span><span class="p">;</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configure</span><span class="p">()</span> <span class="p">{</span>
    <span class="n">setName</span><span class="p">(</span><span class="s">&quot;LogAnalyticsApp&quot;</span><span class="p">);</span>
    <span class="n">setDescription</span><span class="p">(</span><span class="s">&quot;An application that computes the top 10 Client IPs from Apache access log data&quot;</span><span class="p">);</span>
    <span class="n">addStream</span><span class="p">(</span><span class="k">new</span> <span class="n">Stream</span><span class="p">(</span><span class="s">&quot;logEvents&quot;</span><span class="p">));</span>
    <span class="n">addMapReduce</span><span class="p">(</span><span class="k">new</span> <span class="n">TopClientsMapReduce</span><span class="p">());</span>
    <span class="k">try</span> <span class="p">{</span>
      <span class="n">DatasetProperties</span> <span class="n">props</span> <span class="o">=</span> <span class="n">ObjectStores</span><span class="p">.</span><span class="na">objectStoreProperties</span><span class="p">(</span><span class="n">Types</span><span class="p">.</span><span class="na">listOf</span><span class="p">(</span><span class="n">ClientCount</span><span class="p">.</span><span class="na">class</span><span class="p">),</span>
                                                                   <span class="n">DatasetProperties</span><span class="p">.</span><span class="na">EMPTY</span><span class="p">);</span>
      <span class="n">createDataset</span><span class="p">(</span><span class="n">DATASET_NAME</span><span class="p">,</span> <span class="n">ObjectStore</span><span class="p">.</span><span class="na">class</span><span class="p">,</span> <span class="n">props</span><span class="p">);</span>
    <span class="p">}</span> <span class="k">catch</span> <span class="p">(</span><span class="n">UnsupportedTypeException</span> <span class="n">e</span><span class="p">)</span> <span class="p">{</span>
      <span class="k">throw</span> <span class="n">Throwables</span><span class="p">.</span><span class="na">propagate</span><span class="p">(</span><span class="n">e</span><span class="p">);</span>
    <span class="p">}</span>
    <span class="n">addService</span><span class="p">(</span><span class="k">new</span> <span class="n">TopClientsService</span><span class="p">());</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>The <code class="docutils literal notranslate"><span class="pre">LogAnalytics</span></code> application defines a new <a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/streams.html">Stream</a>
where Apache access logs are ingested.</p>
<p>The log events can be ingested into the CDAP stream. Once the data is
ingested, the events can be processed in real-time or batch. In our
application, we will process the events in batch using the
<code class="docutils literal notranslate"><span class="pre">TopClientsMapReduce</span></code> program and compute the top 10 Client IPs in a
specific time-range.</p>
<p>The results of the MapReduce program is persisted into a Dataset; the
application uses the <code class="docutils literal notranslate"><span class="pre">createDataset</span></code> method to define the Dataset.
The <code class="docutils literal notranslate"><span class="pre">ClientCount</span></code> class defines the types used in the Dataset.</p>
<p>Finally, the application adds a service for querying the results from
the Dataset.</p>
<p>Let’s take a closer look at the MapReduce program.</p>
<p>The <code class="docutils literal notranslate"><span class="pre">TopClientsMapReduce</span></code> extends an <a class="reference external" href="http://docs.cdap.io/cdap/current/en/reference-manual/javadocs/co/cask/cdap/api/mapreduce/AbstractMapReduce.html">AbstractMapReduce</a>
class and overrides the <code class="docutils literal notranslate"><span class="pre">configure()</span></code> and <code class="docutils literal notranslate"><span class="pre">initialize()</span></code> methods:</p>
<ul class="simple">
<li><code class="docutils literal notranslate"><span class="pre">configure()</span></code> method configures a MapReduce, setting the program
name, description and output Dataset.</li>
<li><code class="docutils literal notranslate"><span class="pre">initialize()</span></code> method is invoked at runtime, before the MapReduce
is executed. Here you can access the Hadoop job configuration through the
<code class="docutils literal notranslate"><span class="pre">MapReduceContext</span></code> returned by <code class="docutils literal notranslate"><span class="pre">getContext()</span></code>. Mapper, Reducer, and Combiner
classes—as well as the intermediate data format—are set in this method.</li>
</ul>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">TopClientsMapReduce</span> <span class="kd">extends</span> <span class="n">AbstractMapReduce</span> <span class="p">{</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configure</span><span class="p">()</span> <span class="p">{</span>
    <span class="n">setName</span><span class="p">(</span><span class="s">&quot;TopClientsMapReduce&quot;</span><span class="p">);</span>
    <span class="n">setDescription</span><span class="p">(</span><span class="s">&quot;MapReduce program that computes top 10 clients in the last 1 hour&quot;</span><span class="p">);</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">initialize</span><span class="p">()</span> <span class="kd">throws</span> <span class="n">Exception</span> <span class="p">{</span>
    <span class="n">MapReduceContext</span> <span class="n">context</span> <span class="o">=</span> <span class="n">getContext</span><span class="p">();</span>

    <span class="c1">// Get the Hadoop job context, set Mapper, Reducer and Combiner.</span>
    <span class="n">Job</span> <span class="n">job</span> <span class="o">=</span> <span class="p">(</span><span class="n">Job</span><span class="p">)</span> <span class="n">context</span><span class="p">.</span><span class="na">getHadoopJob</span><span class="p">();</span>

    <span class="n">job</span><span class="p">.</span><span class="na">setMapOutputKeyClass</span><span class="p">(</span><span class="n">Text</span><span class="p">.</span><span class="na">class</span><span class="p">);</span>
    <span class="n">job</span><span class="p">.</span><span class="na">setMapOutputValueClass</span><span class="p">(</span><span class="n">IntWritable</span><span class="p">.</span><span class="na">class</span><span class="p">);</span>
    <span class="n">job</span><span class="p">.</span><span class="na">setMapperClass</span><span class="p">(</span><span class="n">IPMapper</span><span class="p">.</span><span class="na">class</span><span class="p">);</span>

    <span class="n">job</span><span class="p">.</span><span class="na">setCombinerClass</span><span class="p">(</span><span class="n">CountsCombiner</span><span class="p">.</span><span class="na">class</span><span class="p">);</span>

    <span class="c1">// Number of reducer set to 1 to compute topN in a single reducer.</span>
    <span class="n">job</span><span class="p">.</span><span class="na">setNumReduceTasks</span><span class="p">(</span><span class="mi">1</span><span class="p">);</span>
    <span class="n">job</span><span class="p">.</span><span class="na">setReducerClass</span><span class="p">(</span><span class="n">TopNClientsReducer</span><span class="p">.</span><span class="na">class</span><span class="p">);</span>

    <span class="c1">// Read events from last 60 minutes as input to the mapper.</span>
    <span class="kd">final</span> <span class="kt">long</span> <span class="n">endTime</span> <span class="o">=</span> <span class="n">context</span><span class="p">.</span><span class="na">getLogicalStartTime</span><span class="p">();</span>
    <span class="kd">final</span> <span class="kt">long</span> <span class="n">startTime</span> <span class="o">=</span> <span class="n">endTime</span> <span class="o">-</span> <span class="n">TimeUnit</span><span class="p">.</span><span class="na">MINUTES</span><span class="p">.</span><span class="na">toMillis</span><span class="p">(</span><span class="mi">60</span><span class="p">);</span>
    <span class="n">context</span><span class="p">.</span><span class="na">addInput</span><span class="p">(</span><span class="n">Input</span><span class="p">.</span><span class="na">ofStream</span><span class="p">(</span><span class="s">&quot;logEvents&quot;</span><span class="p">,</span> <span class="n">startTime</span><span class="p">,</span> <span class="n">endTime</span><span class="p">));</span>
    <span class="n">context</span><span class="p">.</span><span class="na">addOutput</span><span class="p">(</span><span class="n">Output</span><span class="p">.</span><span class="na">ofDataset</span><span class="p">(</span><span class="n">LogAnalyticsApp</span><span class="p">.</span><span class="na">RESULTS_DATASET_NAME</span><span class="p">));</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>In this example, Mapper and Reducer classes are built by implementing
the <a class="reference external" href="http://hadoop.apache.org/docs/r2.3.0/api/org/apache/hadoop/mapreduce/package-summary.html">Hadoop APIs</a>.</p>
<p>In the application, the Mapper class reads the Apache access log event
from the Stream and produces the Client IP and count as the intermediate
map output key and value:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">IPMapper</span> <span class="kd">extends</span> <span class="n">Mapper</span><span class="o">&lt;</span><span class="n">LongWritable</span><span class="p">,</span> <span class="n">Text</span><span class="p">,</span> <span class="n">Text</span><span class="p">,</span> <span class="n">IntWritable</span><span class="o">&gt;</span> <span class="p">{</span>
  <span class="kd">private</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">IntWritable</span> <span class="n">OUTPUT_VALUE</span> <span class="o">=</span> <span class="k">new</span> <span class="n">IntWritable</span><span class="p">(</span><span class="mi">1</span><span class="p">);</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">map</span><span class="p">(</span><span class="n">LongWritable</span> <span class="n">key</span><span class="p">,</span> <span class="n">Text</span> <span class="n">value</span><span class="p">,</span> <span class="n">Context</span> <span class="n">context</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">IOException</span><span class="p">,</span> <span class="n">InterruptedException</span> <span class="p">{</span>
    <span class="c1">// The body of the stream event is contained in the Text value</span>
    <span class="n">String</span> <span class="n">streamBody</span> <span class="o">=</span> <span class="n">value</span><span class="p">.</span><span class="na">toString</span><span class="p">();</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">streamBody</span> <span class="o">!=</span> <span class="kc">null</span>  <span class="o">&amp;&amp;</span> <span class="o">!</span><span class="n">streamBody</span><span class="p">.</span><span class="na">isEmpty</span><span class="p">())</span> <span class="p">{</span>
      <span class="n">String</span> <span class="n">ip</span> <span class="o">=</span> <span class="n">streamBody</span><span class="p">.</span><span class="na">substring</span><span class="p">(</span><span class="mi">0</span><span class="p">,</span> <span class="n">streamBody</span><span class="p">.</span><span class="na">indexOf</span><span class="p">(</span><span class="s">&quot; &quot;</span><span class="p">));</span>
      <span class="c1">// Map output Key: IP and Value: Count</span>
      <span class="n">context</span><span class="p">.</span><span class="na">write</span><span class="p">(</span><span class="k">new</span> <span class="n">Text</span><span class="p">(</span><span class="n">ip</span><span class="p">),</span> <span class="n">OUTPUT_VALUE</span><span class="p">);</span>
    <span class="p">}</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>The reducer class gets the Client IP and count from the MapReducer job and then
aggregates the count for each Client IP and stores it in a priority
queue. The number of reducers is set to 1, so that all results go into
the same reducer to compute the top 10 results. The top 10 results are
written to the MapReduce context in the cleanup method of the Reducer,
which is called once during the end of the task. Writing the results in
the context automatically writes the result to the output Dataset,
specified in the <code class="docutils literal notranslate"><span class="pre">configure()</span></code> method of the MapReduce program.</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">TopNClientsReducer</span> <span class="kd">extends</span> <span class="n">Reducer</span><span class="o">&lt;</span><span class="n">Text</span><span class="p">,</span> <span class="n">IntWritable</span><span class="p">,</span> <span class="kt">byte</span><span class="o">[]</span><span class="p">,</span> <span class="n">List</span><span class="o">&lt;</span><span class="n">ClientCount</span><span class="o">&gt;&gt;</span> <span class="p">{</span>

  <span class="kd">private</span> <span class="kd">static</span> <span class="kd">final</span> <span class="kt">int</span> <span class="n">COUNT</span> <span class="o">=</span> <span class="mi">10</span><span class="p">;</span>
  <span class="kd">private</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">PriorityQueue</span><span class="o">&lt;</span><span class="n">ClientCount</span><span class="o">&gt;</span> <span class="n">priorityQueue</span> <span class="o">=</span> <span class="k">new</span> <span class="n">PriorityQueue</span><span class="o">&lt;</span><span class="n">ClientCount</span><span class="o">&gt;</span><span class="p">(</span><span class="n">COUNT</span><span class="p">);</span>

  <span class="nd">@Override</span>
  <span class="kd">protected</span> <span class="kt">void</span> <span class="nf">reduce</span><span class="p">(</span><span class="n">Text</span> <span class="n">key</span><span class="p">,</span> <span class="n">Iterable</span><span class="o">&lt;</span><span class="n">IntWritable</span><span class="o">&gt;</span> <span class="n">values</span><span class="p">,</span> <span class="n">Context</span> <span class="n">context</span><span class="p">)</span>
                        <span class="kd">throws</span> <span class="n">IOException</span><span class="p">,</span> <span class="n">InterruptedException</span> <span class="p">{</span>
    <span class="c1">// For each Key: IP, aggregate the Value: Count.</span>
    <span class="kt">int</span> <span class="n">count</span> <span class="o">=</span> <span class="mi">0</span><span class="p">;</span>
    <span class="k">for</span> <span class="p">(</span><span class="n">IntWritable</span> <span class="n">data</span> <span class="p">:</span> <span class="n">values</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">count</span> <span class="o">+=</span> <span class="n">data</span><span class="p">.</span><span class="na">get</span><span class="p">();</span>
    <span class="p">}</span>

    <span class="c1">// Store the Key and Value in a priority queue.</span>
    <span class="n">priorityQueue</span><span class="p">.</span><span class="na">add</span><span class="p">(</span><span class="k">new</span> <span class="n">ClientCount</span><span class="p">(</span><span class="n">key</span><span class="p">.</span><span class="na">toString</span><span class="p">(),</span> <span class="n">count</span><span class="p">));</span>

    <span class="c1">// Ensure the priority queue is always contains topN count.</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">priorityQueue</span><span class="p">.</span><span class="na">size</span><span class="p">()</span> <span class="o">&gt;</span> <span class="n">COUNT</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">priorityQueue</span><span class="p">.</span><span class="na">poll</span><span class="p">();</span>
    <span class="p">}</span>
  <span class="p">}</span>

  <span class="nd">@Override</span>
  <span class="kd">protected</span> <span class="kt">void</span> <span class="nf">cleanup</span><span class="p">(</span><span class="n">Context</span> <span class="n">context</span><span class="p">)</span> <span class="kd">throws</span> <span class="n">IOException</span><span class="p">,</span> <span class="n">InterruptedException</span> <span class="p">{</span>
    <span class="c1">// Write topN results in reduce output. Since the &quot;topN&quot; (ObjectStore) Dataset is used as</span>
    <span class="c1">// output the entries will be written to the Dataset without any additional effort.</span>
    <span class="n">List</span><span class="o">&lt;</span><span class="n">ClientCount</span><span class="o">&gt;</span> <span class="n">topNResults</span> <span class="o">=</span> <span class="n">Lists</span><span class="p">.</span><span class="na">newArrayList</span><span class="p">();</span>
    <span class="k">while</span> <span class="p">(</span><span class="n">priorityQueue</span><span class="p">.</span><span class="na">size</span><span class="p">()</span> <span class="o">!=</span> <span class="mi">0</span><span class="p">)</span> <span class="p">{</span>
      <span class="n">topNResults</span><span class="p">.</span><span class="na">add</span><span class="p">(</span><span class="n">priorityQueue</span><span class="p">.</span><span class="na">poll</span><span class="p">());</span>
    <span class="p">}</span>
    <span class="n">context</span><span class="p">.</span><span class="na">write</span><span class="p">(</span><span class="n">TopClientsService</span><span class="p">.</span><span class="na">DATASET_RESULTS_KEY</span><span class="p">,</span> <span class="n">topNResults</span><span class="p">);</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>Now that we have set the data ingestion and processing components, the
next step is to create a service to query the processed data.</p>
<p>The <code class="docutils literal notranslate"><span class="pre">TopClientsService</span></code> defines a simple HTTP RESTful endpoint to perform
this query and return a response:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">TopClientsService</span> <span class="kd">extends</span> <span class="n">AbstractService</span> <span class="p">{</span>

  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="kt">byte</span> <span class="o">[]</span> <span class="n">DATASET_RESULTS_KEY</span> <span class="o">=</span> <span class="p">{</span><span class="sc">&#39;r&#39;</span><span class="p">};</span>

  <span class="nd">@Override</span>
  <span class="kd">protected</span> <span class="kt">void</span> <span class="nf">configure</span><span class="p">()</span> <span class="p">{</span>
    <span class="n">setName</span><span class="p">(</span><span class="s">&quot;TopClientsService&quot;</span><span class="p">);</span>
    <span class="n">addHandler</span><span class="p">(</span><span class="k">new</span> <span class="n">ResultsHandler</span><span class="p">());</span>
  <span class="p">}</span>

  <span class="kd">public</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">ResultsHandler</span> <span class="kd">extends</span> <span class="n">AbstractHttpServiceHandler</span> <span class="p">{</span>

    <span class="nd">@UseDataSet</span><span class="p">(</span><span class="n">LogAnalyticsApp</span><span class="p">.</span><span class="na">DATASET_NAME</span><span class="p">)</span>
    <span class="kd">private</span> <span class="n">ObjectStore</span><span class="o">&lt;</span><span class="n">List</span><span class="o">&lt;</span><span class="n">ClientCount</span><span class="o">&gt;&gt;</span> <span class="n">topN</span><span class="p">;</span>

    <span class="nd">@GET</span>
    <span class="nd">@Path</span><span class="p">(</span><span class="s">&quot;/results&quot;</span><span class="p">)</span>
    <span class="kd">public</span> <span class="kt">void</span> <span class="nf">getResults</span><span class="p">(</span><span class="n">HttpServiceRequest</span> <span class="n">request</span><span class="p">,</span> <span class="n">HttpServiceResponder</span> <span class="n">responder</span><span class="p">)</span> <span class="p">{</span>

      <span class="n">List</span><span class="o">&lt;</span><span class="n">ClientCount</span><span class="o">&gt;</span> <span class="n">result</span> <span class="o">=</span> <span class="n">topN</span><span class="p">.</span><span class="na">read</span><span class="p">(</span><span class="n">DATASET_RESULTS_KEY</span><span class="p">);</span>
      <span class="k">if</span> <span class="p">(</span><span class="n">result</span> <span class="o">==</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
        <span class="n">responder</span><span class="p">.</span><span class="na">sendError</span><span class="p">(</span><span class="mi">404</span><span class="p">,</span> <span class="s">&quot;Result not found&quot;</span><span class="p">);</span>
      <span class="p">}</span> <span class="k">else</span> <span class="p">{</span>
        <span class="n">responder</span><span class="p">.</span><span class="na">sendJson</span><span class="p">(</span><span class="mi">200</span><span class="p">,</span> <span class="n">result</span><span class="p">);</span>
      <span class="p">}</span>
    <span class="p">}</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
</div>
</div>
<div class="section" id="build-and-run-application">
<h2>Build and Run Application<a class="headerlink" href="#build-and-run-application" title="Permalink to this headline">🔗</a></h2>
<p>The <code class="docutils literal notranslate"><span class="pre">LogAnalyticsApp</span></code> can be built and packaged using the Apache Maven command:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> mvn clean package
</pre></div>
</div>
<p>Note that the remaining commands assume that the <code class="docutils literal notranslate"><span class="pre">cdap-cli.sh</span></code> script is
available on your PATH. If this is not the case, please add it:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> <span class="nb">export</span> <span class="nv">PATH</span><span class="o">=</span><span class="nv">$PATH</span>:&lt;CDAP home&gt;/bin
</pre></div>
</div>
<p>If you haven’t already started a standalone CDAP installation, start it with the command:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap.sh start
</pre></div>
</div>
<p>We can then deploy the application to the standalone CDAP installation:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap-cli.sh load artifact target/cdap-mapreduce-guide-&lt;version&gt;.jar
<span class="gp">$</span> cdap-cli.sh create app LogAnalyticsApp cdap-mapreduce-guide &lt;version&gt; user
</pre></div>
</div>
<p>Next, we will send some sample Apache access log event into the stream
for processing:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap-cli.sh send stream logEvents <span class="se">\&#39;</span><span class="m">255</span>.255.255.185 - - <span class="o">[</span><span class="m">23</span>/Sep/2014:11:45:38 -0400<span class="o">]</span> <span class="s2">&quot;GET /cdap.html HTTP/1.0&quot;</span> <span class="m">200</span> <span class="m">190</span> <span class="s2">&quot;Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)&quot;</span><span class="se">\&#39;</span>
<span class="gp">$</span> cdap-cli.sh send stream logEvents <span class="se">\&#39;</span><span class="m">255</span>.255.255.185 - - <span class="o">[</span><span class="m">23</span>/Sep/2014:11:45:38 -0400<span class="o">]</span> <span class="s2">&quot;GET /tigon.html HTTP/1.0&quot;</span> <span class="m">200</span> <span class="m">102</span> <span class="s2">&quot;Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)&quot;</span><span class="se">\&#39;</span>
<span class="gp">$</span> cdap-cli.sh send stream logEvents <span class="se">\&#39;</span><span class="m">255</span>.255.255.185 - - <span class="o">[</span><span class="m">23</span>/Sep/2014:11:45:38 -0400<span class="o">]</span> <span class="s2">&quot;GET /coopr.html HTTP/1.0&quot;</span> <span class="m">200</span> <span class="m">121</span> <span class="s2">&quot;Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)&quot;</span><span class="se">\&#39;</span>
<span class="gp">$</span> cdap-cli.sh send stream logEvents <span class="se">\&#39;</span><span class="m">255</span>.255.255.182 - - <span class="o">[</span><span class="m">23</span>/Sep/2014:11:45:38 -0400<span class="o">]</span> <span class="s2">&quot;GET /tigon.html HTTP/1.0&quot;</span> <span class="m">200</span> <span class="m">111</span> <span class="s2">&quot;Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)&quot;</span><span class="se">\&#39;</span>
<span class="gp">$</span> cdap-cli.sh send stream logEvents <span class="se">\&#39;</span><span class="m">255</span>.255.255.182 - - <span class="o">[</span><span class="m">23</span>/Sep/2014:11:45:38 -0400<span class="o">]</span> <span class="s2">&quot;GET /tigon.html HTTP/1.0&quot;</span> <span class="m">200</span> <span class="m">145</span> <span class="s2">&quot;Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1)&quot;</span><span class="se">\&#39;</span>
</pre></div>
</div>
<p>We can now start the MapReduce program to process the events that were
ingested:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap-cli.sh start mapreduce LogAnalyticsApp.TopClientsMapReduce
</pre></div>
</div>
<p>The MapReduce program will take a couple of moments to process.</p>
<p>We can then start the <code class="docutils literal notranslate"><span class="pre">TopClientsService</span></code> and then query the processed
results:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap-cli.sh start service LogAnalyticsApp.TopClientsService

<span class="gp">$</span> curl -w<span class="s1">&#39;\n&#39;</span> http://localhost:10000/v3/namespaces/default/apps/LogAnalyticsApp/services/TopClientsService/methods/results
</pre></div>
</div>
<p>Example output:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="go">[{&quot;clientIP&quot;:&quot;255.255.255.185&quot;,&quot;count&quot;:3},{&quot;clientIP&quot;:&quot;255.255.255.182&quot;,&quot;count&quot;:2}]</span>
</pre></div>
</div>
<p>You have now learned how to write a MapReduce program to process events from
a Stream, write the results to a Dataset and query the results using a Service.</p>
</div>
<div class="section" id="related-topics">
<h2>Related Topics<a class="headerlink" href="#related-topics" title="Permalink to this headline">🔗</a></h2>
<ul class="simple">
<li><a class="reference external" href="http://docs.cask.co/tutorial/current/en/tutorial2.html">Wise: Web Analytics</a> tutorial, part of CDAP</li>
</ul>
</div>
<div class="section" id="extend-this-example">
<h2>Extend This Example<a class="headerlink" href="#extend-this-example" title="Permalink to this headline">🔗</a></h2>
<p>Now that you have the basics of MapReduce programs down, you can extend
this example by:</p>
<ul class="simple">
<li>Writing a <a class="reference external" href="http://docs.cask.co/cdap/current/en/developers-manual/building-blocks/workflows.html">workflow</a>
to schedule this MapReduce every hour and process the previous hour’s data</li>
<li>Store the results in a Timeseries data to analyze trends</li>
</ul>
</div>
<div class="section" id="share-and-discuss">
<h2>Share and Discuss!<a class="headerlink" href="#share-and-discuss" title="Permalink to this headline">🔗</a></h2>
<p>Have a question? Discuss at the <a class="reference external" href="https://groups.google.com/forum/#!forum/cdap-user">CDAP User Mailing List</a>.</p>
</div>
<div class="section" id="license">
<h2>License<a class="headerlink" href="#license" title="Permalink to this headline">🔗</a></h2>
<p>Copyright © 2014-2015 Cask Data, Inc.</p>
<p>Licensed under the Apache License, Version 2.0 (the “License”); you may
not use this file except in compliance with the License. You may obtain
a copy of the License at</p>
<p><a class="reference external" href="http://www.apache.org/licenses/LICENSE-2.0">http://www.apache.org/licenses/LICENSE-2.0</a></p>
<p>Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an “AS IS” BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.</p>
</div>
</div>

</div>
    <div class="col-md-2">
      <div id="right-sidebar" class="bs-sidenav scrollable-y" role="complementary">
        <div id="localtoc-scrollspy">
        </div>
      </div>
    </div></div>
</div>
<!-- block main content end -->
<!-- block footer -->
<footer class="footer">
      <div class="container">
        <div class="row">
          <div class="col-md-2 footer-left"><a title="Consuming Data from Kafka" href="cdap-kafka-ingest-guide.html" />Previous</a></div>
          <div class="col-md-8 footer-center"><a class="footer-tab-link" href="../table-of-contents/../../reference-manual/licenses/index.html">Copyright</a> &copy; 2014-2020 Cask Data, Inc.&bull; <a class="footer-tab-link" href="//docs.cask.co/cdap/6.1.1/cdap-docs-6.1.1-web.zip" rel="nofollow">Download</a> an archive or
<a class="footer-tab-link" href="//docs.cask.co/cdap">switch the version</a> of the documentation
          </div>
          <div class="col-md-2 footer-right"><a title="Iterative Data Processing with Apache Spark" href="cdap-spark-guide.html" />Next</a></div>
        </div>
      </div>
    </footer>
<!-- block footer end -->
<script type="text/javascript" src="../_static/bootstrap-3.3.6/js/bootstrap.min.js"></script><script type="text/javascript" src="../_static/js/bootstrap-sphinx.js"></script><script type="text/javascript" src="../_static/js/abixTreeList-2.js"></script><script type="text/javascript" src="../_static/js/cdap-dynamicscrollspy-4.js"></script><script type="text/javascript" src="../_static/js/cdap-version-menu.js"></script><script type="text/javascript" src="../_static/js/copy-to-clipboard.js"></script><script type="text/javascript" src="../_static/js/jquery.mousewheel.min.js"></script><script type="text/javascript" src="../_static/js/jquery.mCustomScrollbar.js"></script><script type="text/javascript" src="../_static/js/js.cookie.js"></script><script type="text/javascript" src="../_static/js/tabbed-parsed-literal-0.2.js"></script><script type="text/javascript" src="../_static/js/cdap-onload-javascript.js"></script><script type="text/javascript" src="../_static/js/cdap-version-menu.js"></script>
    <script src="https://cdap.gitee.io/docs/cdap/json-versions.js"/></script>
  </body>
</html>